-module(server).
-behaviour(gen_server).

-export([start_link/0, new_mapper/3, new_reducer/4]).
-export([init/1, handle_cast/2, terminate/2]).

start_link() ->
    gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).

init(_Args) ->
    NodeList = [a@cloud, b@cloud, c@cloud, d@cloud],
    {ok, NodeList}.

handle_cast({newjob, Mapper, Reducer, Input, Output}, State) ->

    NodeList = State,
    FileList = hdfs:get_file_split(Input),
    Reducer_list = new_reducer(Reducer, Output, 1, NodeList),
    _MapperList = new_mapper(FileList, Mapper, Reducer_list),
    receive
    	{ok, code} ->
    	    {stop, normal, State}
    after
	1000 ->
	    {stop, normal, State}		    
    end.

new_mapper([], _Mapper, _Reducer_list) ->
    [];
new_mapper(File_split, Mapper, Reducer_list) ->
    [{File_path, Start, End, Node}|Last] = File_split,
    [spawn(Node, map_server, map_func, [File_path, Start, End, Mapper, Reducer_list])|new_mapper(Last, Mapper, Reducer_list)].

new_reducer(_Reducer, _Output, N, []) ->
    [];
new_reducer(Reducer, Output, N, L) ->
    [H|T] = L,
    Reducer_Pid = spawn(H, reduce_server, reduce_service, [N, Reducer, Output]),
    [{N, Reducer_Pid}|new_reducer(Reducer, Output, N+1, T)].

terminate(Reason, State) ->
    ok.
